package net.bwie.jtp.dwd.log.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class LogSplitProcessFunction extends ProcessFunction<String, String> {

    private final OutputTag<String> errorTag;
    private final OutputTag<String> startTag;
    private final OutputTag<String> displayTag;
    private final OutputTag<String> actionTag;
//
    public LogSplitProcessFunction(OutputTag<String> errorTag, OutputTag<String> startTag,
                                   OutputTag<String> displayTag, OutputTag<String> actionTag){
        this.errorTag = errorTag;
        this.startTag = startTag;
        this.displayTag = displayTag;
        this.actionTag = actionTag;
    }


    @Override
    public void processElement(String value,Context ctx, Collector<String> out) throws Exception {
        JSONObject jsonObject = JSON.parseObject(value);
        if(null!=(jsonObject.get("err"))){
            ctx.output(errorTag,value);
            jsonObject.remove("err");
        }

        if(null!=(jsonObject.get("start"))){
            ctx.output(startTag,jsonObject.toJSONString());
        }
        else{
            //获取曝光和动作日志
            JSONArray displaysJsonArrawy = jsonObject.getJSONArray("displays");
            JSONArray actionsJsonArray = jsonObject.getJSONArray("actions");


            //删除曝光日志和动作日志
            jsonObject.remove("displays");
            jsonObject.remove("actions");


            //输出页面日志（只有common page ts）
            out.collect(jsonObject.toJSONString());

            if(null != displaysJsonArrawy && !displaysJsonArrawy.isEmpty()){
                for (int i=0;i<displaysJsonArrawy.size(); i ++){
                    JSONObject displaysjsonObject= displaysJsonArrawy.getJSONObject(i);
                    jsonObject.put("display",displaysjsonObject);
                    ctx.output(displayTag,jsonObject.toJSONString());
                }
            }
            jsonObject.remove("display");


            if(null!=actionsJsonArray && !actionsJsonArray.isEmpty()){
                for (int i=0;i<actionsJsonArray.size();i++){
                    JSONObject actionsJsonobject = actionsJsonArray.getJSONObject(i);
                    jsonObject.put("action",actionsJsonobject);
                    ctx.output(actionTag,jsonObject.toJSONString());
                }
            }
        }

    }
}
